/**
 * Copyright (C), 2020-2020, 软件公司
 * FileName: RabbitConsumer.java
 * Author:   cakin
 * Date:     2020/5/2
 * Description: 消费端
 */
package com.rabbitmq.simple;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP;


import com.rabbitmq.common.Constant;
import com.rabbitmq.util.ConnectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 消费者
 */
@Slf4j
public class RabbitConsumer {
    /**
     * 消息队列
     * 不需要在RabbitMQ中创建
     * 因为生产者已经帮我们创建了队列
     */
    private static final String QUEUE_NAME = "queue_demo2";

    /**
     * IP地址
     */
    private static final String IP_ADDRESS = "127.0.0.1";

    /**
     * 端口
     */
    private static final int PORT = 5672;

    public static void main( String[] args )  {
        Address[] addresses = new Address[]{
                new Address(IP_ADDRESS, PORT)
        };

        ConnectionFactory factory = new ConnectionFactory();
        //factory.setVirtualHost("/vhost_xyn");
        factory.setUsername("root");
        factory.setPassword("123456");
        try {
            Connection connection = factory.newConnection(addresses);
            final Channel channel = connection.createChannel();
            // 设置客户端最多未被ack的个数
            channel.basicQos(Constant.NUM64);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery( String consumerTag,
                                            Envelope envelope,
                                            AMQP.BasicProperties properties,
                                            byte[] body )
                        throws IOException {
                    log.info(" recv message: " + new String(body));
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        log.info("catch InterruptedException");
                    }
                    /**
                     * deliveryTag（唯一标识 ID）：当一个消费者向 RabbitMQ 注册后，会建立起一个 Channel ，
                     * RabbitMQ 会用 basic.deliver 方法向消费者推送消息，这个方法携带了一个 delivery tag，
                     * 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，
                     * delivery tag 的范围仅限于 Channel
                     * multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，
                     * 则可以一次性确认 delivery_tag 小于等于传入值的所有消息
                     *
                     */
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            /**
             * callback:设置消费者的回调函数。用来处理RabbitMQ推送过来的消息
             * 比如DefaultConsumer，使用时需要重写其中的方法
             */
            channel.basicConsume(QUEUE_NAME, consumer);
            // 等待回调函数执行完毕之后 关闭资源
            TimeUnit.SECONDS.sleep(Constant.NUM5);
            ConnectionUtils.close(channel, connection);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
